package com.lzq.mq.client.consumer;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.constants.NameServerConstants;
import com.lzq.mq.common.dto.*;
import com.lzq.mq.common.enums.*;
import com.lzq.mq.common.remote.BrokerNettyRemoteClient;
import com.lzq.mq.common.remote.NameServerNettyRemoteClient;
import com.lzq.mq.common.utils.AssertUtils;
import com.lzq.mq.common.utils.CollectionUtils;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
public class DefaultConsumer {

    private final static int EACH_BATCH_PULL_MSG_INTER = 100; //如果broker有数据，每间隔100ms拉一批
    private final static int EACH_BATCH_PULL_MSG_INTER_WHEN_NO_MSG = 1000; //如果broker无数据，每间隔1s拉一批

    /**
     * nameserver的ip,端口,用户名,密码
     */
    @Getter
    @Setter
    private String nsIp;

    @Getter
    @Setter
    private Integer nsPort;

    @Getter
    @Setter
    private String nsUser;

    @Getter
    @Setter
    private String nsPwd;

    @Getter
    private NameServerNettyRemoteClient nameServerNettyRemoteClient;

    /**
     * 消费者属性
     */
    @Getter
    @Setter
    private String topic;

    @Getter
    @Setter
    private String consumeGroup;

    @Getter
    @Setter
    private String brokerRole;

    @Getter
    @Setter
    private String brokerGroup;

    @Getter
    @Setter
    private Integer queueId;        // 指定消费的队列(非必要)

    @Getter
    @Setter
    private Integer batchSize;      // 批量拉取数量

    /**
     * broker节点
     */
    @Getter
    @Setter
    private List<String> brokerAddressList;

    /**
     * 消费监听器
     */
    @Getter
    @Setter
    private MessageConsumeListener messageConsumeListener;

    /**
     * broker节点和连接客户端工具
     */
    @Getter
    private final Map<String, BrokerNettyRemoteClient> brokerNettyRemoteClientMap = new ConcurrentHashMap<>();

    public void start() {

        nameServerNettyRemoteClient = new NameServerNettyRemoteClient(nsIp, nsPort);
        nameServerNettyRemoteClient.buildConnection(); // 建立连接
        // 注册服务
        if (!doRegistry()) {
            return;
        }
        // 开启心跳
        startHeartBeatTask();
        // 拉取broker节点
        log.info("开始拉取broker节点...");
        this.fetchBrokerAddress();
        // 连接到broker节点
        //this.connectBroker();
        // 开启消费任务
        //this.startConsumeMsgTask();
    }

    /**
     * 在nameserver注册producer, 使其能够拉取服务
     * @return true:注册成功
     */
    private boolean doRegistry() {
        String registryMsgId = UUID.randomUUID().toString();
        ServiceRegistryReqDTO serviceRegistryReqDTO = new ServiceRegistryReqDTO();
        serviceRegistryReqDTO.setMsgId(registryMsgId);
        serviceRegistryReqDTO.setUser(nsUser);
        serviceRegistryReqDTO.setPassword(nsPwd);
        serviceRegistryReqDTO.setRegistryType(RegistryTypeEnum.CONSUMER.getCode());
        // 设置属性
        TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.REGISTRY.getCode(), JSON.toJSONBytes(serviceRegistryReqDTO));
        // 发送注册消息, 并同步等待
        TcpMsg registryResponse = nameServerNettyRemoteClient.sendSyncMsg(tcpMsg, registryMsgId);
        if (NameServerResponseCode.REGISTRY_SERVICE_SUCCESS.getCode() == registryResponse.getCode()) {
            log.info("注册节点成功!");
            return true;
        } else {
            log.info("注册服务失败");
            return false;
        }
    }

    /**
     * 开启心跳任务
     */
    private void startConsumeMsgTask() {
        Thread consumeTask = new Thread(() -> {
            //不知道对应topic位于哪个Broker节点 todo
            if (BrokerRegistryRoleEnum.SINGLE.getCode().equals(getBrokerRole())) {
                log.info("启动单节点消费...");
                while (true) {
                    try {
                        // log.info("开始拉取消息...");
                        String defaultBrokerAddress = brokerAddressList.get(0);
                        String msgId = UUID.randomUUID().toString();
                        BrokerNettyRemoteClient brokerNettyRemoteClient = this.getBrokerNettyRemoteClientMap().get(defaultBrokerAddress);
                        ConsumeMsgReqDTO consumeMsgReqDTO = new ConsumeMsgReqDTO();
                        consumeMsgReqDTO.setMsgId(msgId);
                        consumeMsgReqDTO.setConsumeGroup(consumeGroup);
                        consumeMsgReqDTO.setTopic(topic);
                        consumeMsgReqDTO.setBatchSize(batchSize);
                        // log.info("发送拉取请求:{}", msgId);
                        TcpMsg pullReqMsg = new TcpMsg(BrokerEventCode.CONSUME_MSG.getCode(), JSON.toJSONBytes(consumeMsgReqDTO));
                        TcpMsg pullMsgResp = brokerNettyRemoteClient.sendSyncMsg(pullReqMsg, msgId, 3, TimeUnit.SECONDS);

                        if (null == pullMsgResp) {
                            log.error("拉取消息失败");
                            TimeUnit.MILLISECONDS.sleep(EACH_BATCH_PULL_MSG_INTER_WHEN_NO_MSG);
                            continue;
                        }

                        ConsumeMsgRespDTO consumeMsgRespDTO = JSON.parseObject(pullMsgResp.getBody(), ConsumeMsgRespDTO.class);
                        List<ConsumeMsgRespEleDTO> consumeMsgRespEleDTOList = consumeMsgRespDTO.getConsumeMsgRespEleDTOList();
                        boolean brokerHasData = false;
                        if(CollectionUtils.isNotEmpty(consumeMsgRespEleDTOList)) {
                            for (ConsumeMsgRespEleDTO consumeMsgRespEleDTO : consumeMsgRespEleDTOList) {
                                List<byte[]> commitLogBodyList = consumeMsgRespEleDTO.getCommitLogContentList();
                                if (CollectionUtils.isEmpty(commitLogBodyList)) {
                                    continue;
                                }
                                brokerHasData = true;
                                List<ConsumeMessage> consumeMessages = new ArrayList<>();
                                for (byte[] bytes : commitLogBodyList) {
                                    ConsumeMessage consumeMessage = new ConsumeMessage();
                                    consumeMessage.setBody(bytes);
                                    consumeMessage.setQueueId(consumeMsgRespEleDTO.getQueueId());
                                    consumeMessages.add(consumeMessage);
                                }
                                ConsumeResult consumeResult = messageConsumeListener.consume(consumeMessages);
                                // 消费成功, 向Broker发送ack
                                if (consumeResult.getConsumeResultStatus() == ConsumeResultStatus.CONSUME_SUCCESS.getCode()) {
                                    String ackMsgId = UUID.randomUUID().toString();
                                    ConsumeMsgAckReqDTO consumeMsgAckReqDTO = new ConsumeMsgAckReqDTO();
                                    consumeMsgAckReqDTO.setAckCount(consumeMessages.size());
                                    consumeMsgAckReqDTO.setConsumeGroup(this.getConsumeGroup());
                                    consumeMsgAckReqDTO.setTopic(this.getTopic());
                                    consumeMsgAckReqDTO.setQueueId(consumeMsgRespEleDTO.getQueueId());
                                    consumeMsgAckReqDTO.setMsgId(ackMsgId);
                                    TcpMsg ackReq = new TcpMsg(BrokerEventCode.CONSUME_SUCCESS_MSG.getCode(), JSON.toJSONBytes(consumeMsgAckReqDTO));
                                    TcpMsg ackResponse = brokerNettyRemoteClient.sendSyncMsg(ackReq, ackMsgId, 3, TimeUnit.SECONDS);
                                    if (null == ackResponse) {
                                        log.error("ack失败!");
                                    }
                                    ConsumeMsgAckRespDTO consumeMsgAckRespDTO = JSON.parseObject(ackResponse.getBody(), ConsumeMsgAckRespDTO.class);
                                    if (AckStatus.SUCCESS.getCode() == consumeMsgAckRespDTO.getAckStatus()) {
                                        log.info("消费ack成功!");
                                    } else {
                                        log.error("消费ack失败!");
                                    }
                                }
                            }
                        }
                        if (brokerHasData) {
                            TimeUnit.MILLISECONDS.sleep(EACH_BATCH_PULL_MSG_INTER);
                        } else {
                            TimeUnit.MILLISECONDS.sleep(EACH_BATCH_PULL_MSG_INTER_WHEN_NO_MSG);
                        }
                    } catch (Exception e) {
                        log.error("消费失败:", e);
                    }
                }
            }
        });
        consumeTask.setName("consume-msg-task");
        consumeTask.start();
    }


    /**
     * 拉取broker节点ip地址
     */
    public void fetchBrokerAddress() {

        String fetchBrokerAddressMsgId = UUID.randomUUID().toString();

        PullBrokerIpReqDTO pullBrokerIpDTO = new PullBrokerIpReqDTO();

        pullBrokerIpDTO.setMsgId(fetchBrokerAddressMsgId);
        pullBrokerIpDTO.setBrokerClusterGroup(brokerGroup);
        pullBrokerIpDTO.setRole(brokerRole);
        TcpMsg req = new TcpMsg(
                NameServerEventCode.PULL_BROKER_IP_LIST.getCode(),
                JSON.toJSONBytes(pullBrokerIpDTO)
        );
        // 发送
        TcpMsg resp = nameServerNettyRemoteClient.sendSyncMsg(
                req,
                fetchBrokerAddressMsgId
        );

        //获取broker节点ip地址，并且缓存起来，可能由多个master-broker角色
        PullBrokerIpRespDTO pullBrokerIpRespDTO = JSON.parseObject(resp.getBody(), PullBrokerIpRespDTO.class);
        setBrokerAddressList(pullBrokerIpRespDTO.getAddressList());
        log.info("拉取broker节点ip地址成功:{}", pullBrokerIpRespDTO);
    }

    /**
     * 开启心跳任务
     */
    private void startHeartBeatTask() {
        Thread heartBeatTask = new Thread(new Runnable() {
            @Override
            public void run() {
                log.info("开启心跳任务");
                while (true) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(NameServerConstants.SERVER_HEART_BEAT_INTERVAL_MS);
                        String heartBeatMsgId = UUID.randomUUID().toString();
                        HeartBeatDTO heartBeatDTO = new HeartBeatDTO();
                        heartBeatDTO.setMsgId(heartBeatMsgId);
                        TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.HEART_BEAT.getCode(),
                                JSON.toJSONBytes(heartBeatDTO));
                        TcpMsg heartBeatResponse = nameServerNettyRemoteClient.sendSyncMsg(tcpMsg, heartBeatMsgId);
                    } catch (InterruptedException e) {
                        log.error("心跳任务异常", e);
                    }
                }
            }
        }, "DefaultProducer心跳任务");
        heartBeatTask.start();
    }

    /**
     * 连接到节点
     */
    private void connectBroker() {
        AssertUtils.isNotEmpty(this.getBrokerAddressList(),"broker地址不能为空");
        for (String brokerIp : brokerAddressList) {
            String[] brokerAddressArr = brokerIp.split(":");
            BrokerNettyRemoteClient brokerNettyRemoteClient = new BrokerNettyRemoteClient(brokerAddressArr[0],
                    Integer.valueOf(brokerAddressArr[1]));
            brokerNettyRemoteClient.buildConnection();
            this.brokerNettyRemoteClientMap.put(brokerIp,brokerNettyRemoteClient);
        }
        log.info("broker连接成功:{}", brokerAddressList);
    }

    /**
     * 获取broker实例
     * @param topic  topic
     * @return broker实例
     */
    private BrokerNettyRemoteClient getRemoteClient(String topic) {

        return this.brokerNettyRemoteClientMap.values().stream().collect(Collectors.toList()).get(0);
    }
}
